#include "rpcprovider.h"
#include "mprpcapplication.h"
#include <functional>
#include <muduo/base/Logging.h>
#include <google/protobuf/descriptor.h>
#include "RpcHeader.pb.h"
#include "zookeeperutil.h"
void RpcProvider::NotifyService(google::protobuf::Service* service)
{
    ServiceInfo service_info;
    const google::protobuf::ServiceDescriptor* sp=service->GetDescriptor();

    std::string service_name=sp->name();
    int method_cnt=sp->method_count();

    for(int i=0;i<method_cnt;i++)
    {
        const google::protobuf::MethodDescriptor* methodp=sp->method(i);
        std::string method_name=methodp->name();
        service_info.m_methodMap.insert({method_name,methodp});
    }

    service_info.service_=service;
    m_serviceInfoMap.insert({service_name,service_info});
}

void RpcProvider::run()
{
    std::string ip=MrpcApplication::getInstance().getConfig().Load("rpcserverip");
    std::string port=MrpcApplication::getInstance().getConfig().Load("rpcserverport");
    LOG_INFO<<"ip : "<<ip<<"\n";
    LOG_INFO<<"port : "<<port<<"\n";
    muduo::net::InetAddress addr(ip,atoi(port.c_str()));

    muduo::net::TcpServer server(&loop_,addr,"Rpcprovider");

    server.setConnectionCallback(std::bind(&RpcProvider::onConnection,this,std::placeholders::_1));
    server.setMessageCallback(std::bind(&RpcProvider::onMessage,this,
                        std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
    
    
    server.setThreadNum(3);
    //SEESION  30S    
    ZkClient zkCli;
    zkCli.start();

    //service_name is permanent method_name is temporary
    for(auto &sp:m_serviceInfoMap)
    {
        std::string service_path="/"+sp.first;
        //PERMANENT
        zkCli.Create(service_path.c_str(),nullptr,0);
        for(auto &mp:sp.second.m_methodMap)
        {
            std::string method_path=service_path+"/"+mp.first;
            char method_path_data[128]={0};
            sprintf(method_path_data,"%s:%d",ip.c_str(),atoi(port.c_str()));
            //TEMPORARY
            zkCli.Create(method_path.c_str(),method_path_data,strlen(method_path_data),ZOO_EPHEMERAL);
        }
    }

    server.start();
    loop_.loop();
}

void RpcProvider::onConnection(const muduo::net::TcpConnectionPtr& conn)
{
    if(!conn->connected())
    {
        conn->shutdown();
        return;
    }
    LOG_INFO<<"CONNECTION IS DONE\n";
}

void RpcProvider::onMessage(const muduo::net::TcpConnectionPtr& conn,muduo::net::Buffer* buffer
                    , muduo::Timestamp time)
{
    std::string recv_buf=buffer->retrieveAllAsString();
    uint32_t header_size=0;
    recv_buf.copy((char*)&header_size,4,0);

    std::string rpc_header_str=recv_buf.substr(4,header_size);
    mprpc::RpcHeader rpcHeader;
    std::string service_name;
    std::string method_name;
    uint32_t args_size=0;
    if(rpcHeader.ParseFromString(rpc_header_str))
    {
        service_name=rpcHeader.service_name();
        method_name=rpcHeader.method_name();
        args_size=rpcHeader.args_size();
    }
    else{
        LOG_ERROR<<"RPCHEADER_STR PARSE ERROR\n";
        return;
    }
    std::string args_str=recv_buf.substr(4+header_size,args_size);
    LOG_INFO<<"RpcProvider::onMessage>>\n";
    LOG_INFO<< time.toString()<<"\n";
    LOG_INFO<<"service_name: "<<service_name<<"\n";
    LOG_INFO<<"method_name:  "<<method_name<<"\n";
    LOG_INFO<<"args_str:  "<<args_str<<"\n";
    
    auto it=m_serviceInfoMap.find(service_name);
    if(it==m_serviceInfoMap.end())
    {
        LOG_ERROR<<"RpcProvider::onMessage>>SERVICE_NAME IS NOT EXISTING\n";
        return;
    }

    google::protobuf::Service* service=it->second.service_;
    auto mit=it->second.m_methodMap.find(method_name);
    if(mit==it->second.m_methodMap.end())
    {
        LOG_ERROR<<"RpcProvider::onMessage>>METHOD_NAME IS NOT EXISTING\n";
        return;
    }

    const google::protobuf::MethodDescriptor* methodp=mit->second;

    google::protobuf::Message* request=service->GetRequestPrototype(methodp).New();
    google::protobuf::Message* response=service->GetResponsePrototype(methodp).New();
    if(!request->ParseFromString(args_str))
    {
        LOG_ERROR<<"REQUEST PARSE ERROR\n";
        return;
    }
    google::protobuf::Closure* done= google::protobuf::NewCallback
    <RpcProvider,const muduo::net::TcpConnectionPtr&,google::protobuf::Message*>
    (this,&RpcProvider::SendResponse,conn,response);

    service->CallMethod(methodp,nullptr,request,response,done );
}

void RpcProvider::SendResponse(const muduo::net::TcpConnectionPtr& conn,google::protobuf::Message* response)
{
    std::string response_str;
    if(response->SerializeToString(&response_str))
    {
        conn->send(response_str);
        
    }
    else{
        LOG_ERROR<<"RpcProvider::SendResponse>>SerializeToString error\n";
    }
    conn->shutdown();
}